package example;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileReader;
import java.io.IOException;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Random;
import java.util.StringTokenizer;
import java.util.concurrent.TimeUnit;

import org.apache.cassandra.thrift.TimedOutException;

import com.yammer.metrics.Metrics;
import com.yammer.metrics.core.Meter;

public class SlaveClientThread extends Thread{
	Connection[] connections;
	PreparedStatement[] statements;
	Random random=new Random(System.nanoTime());
	static int pid = 0;
	Meter meter = Metrics.newMeter(SlaveClientThread.class, "requests" + (pid++),"requests" + + (pid++), TimeUnit.SECONDS);
	SimpleClient father;
	public SlaveClientThread(String[] ips,String port,SimpleClient faClient) throws SQLException{
		connections=new Connection[ips.length];
		for(int i=0;i<ips.length;i++){
			System.out.println("slave connect:"+"jdbc:cassandra://"+ips[i]+":"+port);
			connections[i]=DriverManager.getConnection("jdbc:cassandra://"+ips[i]+":"+port);
			connections[i].createStatement().execute("use test_metrics"+SimpleClient.keyspace );
		}
		statements=new PreparedStatement[connections.length];
		this.father=faClient;
	}
	public void run(){
		File file= SimpleClient.requestFile();
		String query = "UPDATE query SET word=?, rank=?,squenceid=?,url=? WHERE uid=? and stime=?";
		PreparedStatement statement=null;
		boolean close=false;
		if(file==null){
			close=true;
		}
		for(int i=0;i<statements.length;i++){
			try {
				statements[i] = connections[i].prepareStatement(query);
			} catch (SQLException e1) {
				e1.printStackTrace();
				close=true;
			}
		}
		if(close){
			for(int i=0;i<connections.length;i++){
				try {
					connections[i].close();
				} catch (SQLException e) {
					e.printStackTrace();
				}
			}
			System.out.println("thread"+getId()+" closed....");
			return;
		}
		BufferedReader brReader;
		try {
			brReader = new BufferedReader(new FileReader(file));
			String line=null;
			String word="";
			int rank=0;
			int squenceid=0;
			String url="";
			String uid="";
			String stime="";
			int i=0;
			int timeMs = 0;
			int timeMsDecreaseIn = 0;
			int n=0;
			int max=0;
			while((line=brReader.readLine())!=null){
				StringTokenizer tokenizer=new StringTokenizer(line);
				if(tokenizer.hasMoreTokens()){
					uid=tokenizer.nextToken();
				}
				if(tokenizer.hasMoreTokens()){
					word=tokenizer.nextToken();
				}
				if(tokenizer.hasMoreTokens()){
					try{
						rank=Integer.valueOf(tokenizer.nextToken());
					}catch(Exception e){	
						System.err.println(line);
						continue;
					}
				}
				if(tokenizer.hasMoreTokens()){
					try{
						squenceid=Integer.valueOf(tokenizer.nextToken());
					}catch(Exception e){	
						System.err.println(line);
						continue;
					}
				}
				if(tokenizer.hasMoreTokens()){
					url=tokenizer.nextToken();
				}
				statement=statements[random.nextInt(statements.length)];
				statement.setString(1, word);
				statement.setInt(2, rank);
				statement.setInt(3, squenceid);
				statement.setString(4, url);
				statement.setString(5, uid);
				statement.setString(6, stime);
				try{
					
					statement.executeUpdate();
					meter.mark();
					if(meter.oneMinuteRate()>max){
						max=(int)meter.oneMinuteRate();
					}
					if(n<=0){
						if(father.rate!=0){
							double time= (meter.oneMinuteRate()-father.rate);
							if(time>0){
								//timeMs=(int)(time/(meter.oneMinuteRate()*father.rate)*1000);
								timeMs = 1000 / father.rate;//如果过快则让线程休息timeMs毫秒
								if(max>=1000){
									n=max/1000;//每隔n次测试一次,simpleK为系数
								}else{
									n=(father.rate - 1) * father.simpleK;//每隔n次测试一次,simpleK为系数
								}
								timeMsDecreaseIn = n;
							} else if (time < 0.01) {
								//每隔timeMsDecreaseIn次，休眠时间减少1秒，这样可以缓慢地将速度恢复到father.rate
								if (timeMsDecreaseIn <= 0) {
									timeMs = Math.max(1, timeMs - 1);
									if(timeMs==1){
										if(max>=1000){
											n=max/1000;//每隔n次测试一次,simpleK为系数
										}else{
											n=(father.rate - 1) * father.simpleK;//每隔n次测试一次,simpleK为系数
										}
									}
								} else {
									timeMsDecreaseIn--;
								}
								
							}
							if(timeMs>0){
								if(father.rate>=1000)
									if(time<0.01)
										Thread.sleep(new Random().nextInt(father.rate)==1?1:0);
									else
										Thread.sleep(new Random().nextInt(father.rate/1000+1)==1?1:0);
								else
									Thread.sleep(timeMs + father.threads.length / 2);
							}else if(timeMs==0){
								//当father。rate过大时，timeMs=0；timeMsDecreaseIn也很大，导致最终每次休息时间都是Thread.sleep(Math.max(1, father.threads.length / 2));
								//这样休息时间太久了。我们这时候按照father.rate/1000来休息，
//								
								if(father.rate>=1000)
									if(time<0.01)
										Thread.sleep(new Random().nextInt(father.rate)==1?1:0);
									else
										Thread.sleep(new Random().nextInt(father.rate/1000+1)==1?1:0);
								else
									Thread.sleep(Math.max(1, father.threads.length / 2));
							}
							if(time>0){
								if((time+0.0)/father.rate<0.01){
									n=1;
								}else{
									n=0;
								}
							}
						} 
					}else{
						n--;
					}
				}catch(Exception e){
					if (e instanceof TimedOutException){
						System.out.println(getId()+" timeout,will sleep 5 seconds...");
						try {
							Thread.sleep(5);
						} catch (InterruptedException e1) {
							e1.printStackTrace();
						}
					}
				}
				i++;
				//				if(i%10==0){
				//					try {
				//						Thread.sleep((int)P_rand(2));
				//					} catch (InterruptedException e) {
				//						e.printStackTrace();
				//					}
				//				}
				//				if(i%100==0)
				//					System.out.println(Thread.currentThread().getId()+" insert "+i +"rows..");
			}
		} catch (FileNotFoundException e) {
			e.printStackTrace();
		} catch (IOException e) {
			e.printStackTrace();
		} catch (SQLException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}finally{
			for(int i=0;i<connections.length;i++){
				try {
					connections[i].close();
				} catch (SQLException e) {
					e.printStackTrace();
				}
			}
		}

	}
	public double getDynamicRate(){
		return meter.oneMinuteRate();
	}
	// 泊松分布
	public double P_rand(double Lamda){      
		double x=0,b=1,c=Math.exp(-Lamda),u; 
		do {
			u=Math.random();
			b *=u;
			if(b>=c)
				x++;
		}while(b>=c);
		return x;
	}
}
